#pragma once

#include <faabric/util/config.h>
#include <faabric/util/exception.h>

#include <hiredis/hiredis.h>
#include <memory>
#include <mutex>
#include <set>
#include <string>
#include <string_view>
#include <thread>
#include <unordered_set>
#include <vector>

namespace faabric::redis {
enum RedisRole
{
    QUEUE,
    STATE,
};

class RedisInstance
{
  public:
    explicit RedisInstance(RedisRole role);

    std::string delifeqSha;
    std::string schedPublishSha;

    std::string ip;
    std::string hostname;
    int port;

  private:
    RedisRole role;

    std::mutex scriptsLock;

    std::string loadScript(redisContext* context,
                           const std::string_view scriptBody);

    // Script to delete a key if it equals a given value
    const std::string_view delifeqCmd = R"---(
if redis.call('GET', KEYS[1]) == ARGV[1] then
    return redis.call('DEL', KEYS[1])
else
    return 0
end
)---";

    // Script to push and expire function execution results avoiding extra
    // copies and round-trips
    const std::string_view schedPublishCmd = R"---(
local key = KEYS[1]
local status_key = KEYS[2]
local result = ARGV[1]
local result_expiry = tonumber(ARGV[2])
local status_expiry = tonumber(ARGV[3])
redis.call('RPUSH', key, result)
redis.call('EXPIRE', key, result_expiry)
redis.call('SET', status_key, result)
redis.call('EXPIRE', status_key, status_expiry)
return 0
)---";
};

using UniqueRedisReply =
  std::unique_ptr<redisReply, decltype(&freeReplyObject)>;

class Redis
{

  public:
    ~Redis();

    /**
     *  ------ Factories ------
     */

    static Redis& getQueue();

    static Redis& getState();

    /**
     *  ------ Standard Redis commands ------
     */
    void ping();

    std::vector<uint8_t> get(const std::string& key);

    size_t strlen(const std::string& key);

    void get(const std::string& key, uint8_t* buffer, size_t size);

    void set(const std::string& key, const std::vector<uint8_t>& value);

    void set(const std::string& key, const uint8_t* value, size_t size);

    void del(const std::string& key);

    long getCounter(const std::string& key);

    long incr(const std::string& key);

    long decr(const std::string& key);

    long incrByLong(const std::string& key, long val);

    long decrByLong(const std::string& key, long val);

    void setRange(const std::string& key,
                  long offset,
                  const uint8_t* value,
                  size_t size);

    void setRangePipeline(const std::string& key,
                          long offset,
                          const uint8_t* value,
                          size_t size);

    void flushPipeline(long pipelineLength);

    void getRange(const std::string& key,
                  uint8_t* buffer,
                  size_t bufferLen,
                  long start,
                  long end);

    void sadd(const std::string& key, const std::string& value);

    void srem(const std::string& key, const std::string& value);

    long scard(const std::string& key);

    bool sismember(const std::string& key, const std::string& value);

    std::string srandmember(const std::string& key);

    std::set<std::string> smembers(const std::string& key);

    std::set<std::string> sdiff(const std::string& keyA,
                                const std::string& keyB);

    std::set<std::string> sinter(const std::string& keyA,
                                 const std::string& keyB);

    int lpushLong(const std::string& key, long val);

    int rpushLong(const std::string& key, long val);

    void flushAll();

    long listLength(const std::string& queueName);

    long getTtl(const std::string& key);

    void expire(const std::string& key, long expiry);

    void refresh();

    /**
     *  ------ Locking ------
     */

    uint32_t acquireLock(const std::string& key, int expirySeconds);

    void releaseLock(const std::string& key, uint32_t lockId);

    void delIfEq(const std::string& key, uint32_t value);

    bool setnxex(const std::string& key, long value, int expirySeconds);

    long getLong(const std::string& key);

    void setLong(const std::string& key, long value);

    /**
     * ------ Queueing ------
     */
    void enqueue(const std::string& queueName, const std::string& value);

    void enqueueBytes(const std::string& queueName,
                      const std::vector<uint8_t>& value);

    void enqueueBytes(const std::string& queueName,
                      const uint8_t* buffer,
                      size_t bufferLen);

    std::string dequeue(const std::string& queueName,
                        int timeout = DEFAULT_TIMEOUT);

    std::vector<uint8_t> dequeueBytes(const std::string& queueName,
                                      int timeout = DEFAULT_TIMEOUT);

    size_t dequeueBytes(const std::string& queueName,
                        uint8_t* buffer,
                        size_t bufferLen,
                        int timeout = DEFAULT_TIMEOUT);

    void dequeueMultiple(const std::string& queueName,
                         uint8_t* buff,
                         long buffLen,
                         long nElems);

    // Scheduler result publish
    void publishSchedulerResult(const std::string& key,
                                const std::string& status_key,
                                const std::vector<uint8_t>& result);

  private:
    explicit Redis(const RedisInstance& instance);

    redisContext* context;

    const RedisInstance& instance;

    UniqueRedisReply dequeueBase(const std::string& queueName, int timeout);
};

class RedisNoResponseException : public faabric::util::FaabricException
{
  public:
    explicit RedisNoResponseException(std::string message)
      : FaabricException(std::move(message))
    {}
};
};
